<?php
/**
 * Redis 5.0 引入了 Stream 数据结构，提供更强大的消息队列功能，支持持久化和消息确认。
 * 更高效的任务队列管理
 * Stream 是一种数据结构
 * 可以是消息广播
 * 可以是消息队列 Consumer Group 机制
 * 基本流程
 * 生产者：使用 XADD 将消息添加到 Stream。
 * 消费者：使用 XREAD 或 XREADGROUP 读取消息。
 *
 *
 * 添加数据 php mq/mqStreamList.php mq
 * 消费数据 php mq/mqStreamList.php
 */

class MqList
{
    /**
     * 添加数据
     */
    static function mq($data = ['task_id' => 1, 'message' => 'Stream Queue Example'])
    {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        $data['task_id'] = date('YmdHis');
        //$d = json_encode($data);
        $d = $data;
        //创建消费者组
        $redis->xGroup('CREATE', 'mystream', 'mygroup', '$', true);    // '*' 表示自动生成ID
        $redis->xAdd('mystream', '*', $d);    // '*' 表示自动生成ID
        echo "Message added to stream.\n";
    }

    /**
     * 消费数据
     */
    static function server()
    {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        # 使用不同的消费者名字
        $consumer = 'consumer:' . getmygid();
        $streamName = 'mystream';
        $groupName = 'mygroup';
        while (true) {
            $messages = $redis->xReadGroup(
                $groupName,    // 组名
                $consumer,  // 消费者名
                [$streamName => '>'],  // '>' 表示读取未被组内消费的消息
                1,          //限制每次处理一条
            );
            if ($messages) {
                //$stream  就是key=mystream
                foreach ($messages as $stream => $entries) {
                    foreach ($entries as $id => $fields) {
                        //处理消息
                        echo "Received message ID {$id}: " . json_encode($fields) . "\n";
                        // 确认消息已处理
                        $redis->xAck($streamName, $groupName, [$id]);
                    }
                }
            }
        }
    }
}

if ('mq' == ($argv[1] ?? '') || 'php-fpm' == PHP_SAPI) {
    MqList::mq();
    return;
}

MqList::server();







